package glue_datasyncer

import (
	"context"
	"errors"
	"fmt"
	"github.com/kataras/iris/v12"
	"github.com/spf13/pflag"
	"github.com/spf13/viper"
	"github.com/wutianze/nats.go"
	streamCtx "gluenet/internal/glue-datasyncer/context"
	db2 "gluenet/internal/glue-datasyncer/db"
	v12 "gluenet/pkg/log/v1"
	"gopkg.in/yaml.v3"
	"io/ioutil"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"net/http"
	"os"
	"path/filepath"
	"sync"
	"time"
)

type StorageServer struct {
	ctx    context.Context
	logger v12.Logger
	DB     *db2.DB

	configMapNameSpace string
	configMapName      string
	kubeConfig         string

	promReciverPort   string
	promReciverServer *http.Server
	serverPort        string
	server            *iris.Application
	nats              *nats.Conn

	promeReloadUrl  string
	promeReloadPath string

	metricSubChan *sync.Map
	metricChan    chan []byte
	metricLabel   *sync.Map

	logSubChan  *sync.Map
	logChan     chan []byte
	logProducer *sync.Map
}

func (r *StorageServer) InitFlag(flags *pflag.FlagSet) {
	flags.String("server.port", "10000", "gluenet server serverPort")
	flags.String("write.port", "10001", "gluenet server serverPort")

	flags.String("configmap.namespace", "default", "gluenet server serverPort")
	flags.String("configmap.name", "prometheus-rules", "gluenet server serverPort")
	flags.String("kube.config", "/config/kubeconfig", "gluenet server serverPort")

	flags.String("prom.url", "localhost:9090", "gluenet server serverPort")
	flags.String("prom.path", "/config/cache/", "gluenet server serverPort")
}

func (r *StorageServer) ViperConfig(viper *viper.Viper) {
}

func (r *StorageServer) InitViper(viper *viper.Viper) {
	r.serverPort = viper.GetString("server.port")
	r.promReciverPort = viper.GetString("write.port")

	r.configMapNameSpace = viper.GetString("configmap.namespace")
	r.configMapName = viper.GetString("configmap.name")
	r.kubeConfig = viper.GetString("kube.config")

	r.promeReloadUrl = viper.GetString("prom.url")
	r.promeReloadPath = viper.GetString("prom.path")
}

func (r *StorageServer) Initialize() {
	var (
		err       error
		k8sClient *kubernetes.Clientset
		configMap *v1.ConfigMap
	)

	r.logger = v12.NewLogger("runner")

	mode := os.Getenv("MODE")
	if mode == "prod" {
		k8sClient, configMap, err = saveRulesFromConfigMapShareCache(r.ctx, r.kubeConfig,
			r.configMapNameSpace, r.configMapName, r.promeReloadPath, r.promeReloadUrl)
		if err != nil {
			r.logger.Fatalf("save config map to sharePath")
		}
	}

	// prometheus remote write
	r.promReciverServer = &http.Server{
		ReadTimeout:  60 * time.Second,
		WriteTimeout: 60 * time.Second,
		Addr:         "0.0.0.0:" + r.promReciverPort,
	}
	r.server = iris.New()
	natsUrl := os.Getenv("NATS_SERVER")
	if natsUrl == "" {
		natsUrl = "nats://39.101.140.145:4222"
		r.logger.Debugf("NATS_SERVER not config")
	}
	r.nats, err = nats.Connect(natsUrl)
	if err != nil {
		r.logger.Fatalf("connect nats %v", err)
	}
	db2.InstallDB()
	r.DB = db2.DefaultDB

	// init context
	sctx := streamCtx.StreamContext{
		Context: r.ctx,
		Logger:  r.logger,

		MetricsChan:   r.metricChan,
		MetricSubChan: r.metricSubChan,
		MetricLabels:  r.metricLabel,
		ReciverServer: r.promReciverServer,
		AppServer:     r.server,
		Nats:          r.nats,
		DB:            r.DB,
		ConfigMapUpdater: func(data map[string]interface{}, file string) error {
			if mode == "prod" {
				bts, _ := yaml.Marshal(data)
				if err := ioutil.WriteFile(filepath.Join(filepath.Dir(r.promeReloadPath), filepath.Base(file)), bts, 0666); err != nil {
					return err
				}
				configMap.Data[filepath.Base(file)] = string(bts)
				configMapNew, err := k8sClient.CoreV1().ConfigMaps(r.configMapNameSpace).Update(r.ctx, configMap, metav1.UpdateOptions{})
				if err != nil {
					return err
				}
				_, err = http.Post(fmt.Sprintf("http://%v/-/reload", r.promeReloadUrl), "", nil)
				if err != nil {
					return err
				}
				configMap = configMapNew
			}
			return nil
		},
		ConfigMapGeter: func(file string) (map[string]interface{}, error) {
			if mode == "prod" {
				bts, err := ioutil.ReadFile(filepath.Join(filepath.Dir(r.promeReloadPath), filepath.Base(file)))
				if err != nil {
					return nil, err
				}
				var data map[string]interface{}
				if err = yaml.Unmarshal([]byte(bts), &data); err != nil {
					return nil, err
				}
				return data, nil
			}
			return nil, nil
		},
	}

	// register handler func
	Register(&sctx)

	// allocate stream
	go MetricsRouter(&sctx)
	go LogsRouter(&sctx)

	go func() {
		if err := r.promReciverServer.ListenAndServe(); err != nil {
			if errors.Is(err, http.ErrServerClosed) {
				r.logger.Infof("server closed")
			} else {
				r.logger.Infof("server error %v", err)
			}
		}
	}()

	go func() {
		if err := r.server.Run(iris.Addr(":" + r.serverPort)); err != nil {
			if errors.Is(err, iris.ErrServerClosed) {
				r.logger.Infof("server closed")
			} else {
				r.logger.Infof("server error %v", err)
			}
		}
	}()

	r.logger.Infof("server initialize finished")
	<-r.ctx.Done()
}

func (r *StorageServer) Close() {
	if err := r.promReciverServer.Close(); err != nil {
		r.logger.Errorf("close glue.impl %v", err)
	}
}

func NewServer(ctx context.Context) *StorageServer {
	r := StorageServer{
		ctx: ctx,

		metricChan:    make(chan []byte),
		metricSubChan: &sync.Map{},
		metricLabel:   &sync.Map{},

		logChan:     make(chan []byte),
		logSubChan:  &sync.Map{},
		logProducer: &sync.Map{},
	}
	return &r
}

func saveRulesFromConfigMapShareCache(ctx context.Context, kubeConfig string, namespace string, configmap string, sharePath string, promUrl string) (*kubernetes.Clientset, *v1.ConfigMap, error) {
	config, err := clientcmd.BuildConfigFromFlags("", kubeConfig)
	if err != nil {
		return nil, nil, err
	}
	clt, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, nil, err
	}
	cfg, err := clt.CoreV1().ConfigMaps(namespace).Get(ctx, configmap, metav1.GetOptions{})
	if err != nil {
		return nil, nil, err
	}

	for fileName, fileValue := range cfg.Data {
		//fmt.Printf("path %v", filepath.Join(sharePath, filepath.Base(fileName)))
		if err = ioutil.WriteFile(filepath.Join(sharePath, filepath.Base(fileName)), []byte(fileValue), 0666); err != nil {
			return nil, nil, err
		}
	}

	http.DefaultClient.Timeout = time.Second * 30
	_, err = http.Post(fmt.Sprintf("http://%v/-/reload", promUrl), "", nil)
	if err != nil {
		return nil, nil, err
	}
	return clt, cfg, nil
}
